package org.budo.canal.message.handler;

import java.util.ArrayList;
import java.util.List;

import org.budo.support.lang.util.StringUtil;

import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;

/**
 * @author limingwei
 */
public class RowChangeUtil {
    public static String rowChangeToString(List<RowData> rowDatas) {
        if (null == rowDatas || rowDatas.isEmpty()) {
            return "rowChange.rowDatas is null";
        }

        List<String> rowDataStringList = new ArrayList<String>();
        for (RowData rowData : rowDatas) {
            String rowDataToString = rowDataToString(rowData);
            rowDataStringList.add(rowDataToString);
        }

        return "rowDatas=[" + StringUtil.join(rowDataStringList) + "]";
    }

    public static String rowChangeToString(RowChange rowChange) {
        if (null == rowChange) {
            return "rowChange is null";
        }

        if (StringUtil.isNotBlank(rowChange.getSql())) {
            return "rowChange.sql=" + rowChange.getSql();
        }

        List<RowData> rowDatas = rowChange.getRowDatasList();

        if (null == rowDatas || rowDatas.isEmpty()) {
            return "rowChange.rowDatas is null, rowChange=" + rowChange;
        }

        return rowChangeToString(rowDatas);
    }

    private static String rowDataToString(RowData rowData) {
        List<Column> beforeColumns = rowData.getBeforeColumnsList();
        List<Column> afterColumns = rowData.getAfterColumnsList();

        return columnsInfo(beforeColumns) + " : " + columnsValue(beforeColumns) + ">" + columnsValue(afterColumns);
    }

    private static String columnsInfo(List<Column> columns) {
        List<String> list = new ArrayList<String>();

        for (Column column : columns) {
            list.add("{\"name\":\"" + column.getName() + "\", \"isKey\"" + column.getIsKey() + "\"}");
        }

        return "[" + StringUtil.join(list) + "]";
    }

    private static String columnsValue(List<Column> columns) {
        List<String> values = new ArrayList<String>();

        for (Column column : columns) {
            column.getIsKey();
            column.hasIsKey();
            values.add(column.getValue());
        }

        return "[" + StringUtil.join(values) + "]";
    }

    public static RowChange rowChangeParseFromByteString(ByteString byteString) {
        try {
            return RowChange.parseFrom(byteString);
        } catch (InvalidProtocolBufferException e) {
            throw new RuntimeException(e);
        }
    }

    public static Integer getIsKeyCount(List<Column> columns) {
        if (null == columns) {
            return null;
        }

        Integer isKeyCount = 0;
        for (Column column : columns) {
            if (column.getIsKey()) {
                isKeyCount++;
            }
        }

        return isKeyCount;
    }

    public static boolean hasIsKeyColumn(List<Column> columns) {
        if (null == columns) {
            return false;
        }

        for (Column column : columns) {
            if (column.getIsKey()) {
                return true;
            }
        }

        return false;
    }
}